package com.caul.demo.hadoop.mapreduce.ext;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * Created by BlueDream on 2018-04-11.
 */
public class CustomOutputFormat extends FileOutputFormat<Text, CustomWritable> {


  @Override
  public RecordWriter<Text, CustomWritable> getRecordWriter(TaskAttemptContext context)
      throws IOException, InterruptedException {
    FileSystem fs = FileSystem.get(context.getConfiguration());
    FSDataOutputStream output = fs.create(new Path("hdfs://cdh-namenode:9000/out/custom.dat"));
    return new CustomRecordWriter(output);
  }

  private class CustomRecordWriter extends RecordWriter<Text, CustomWritable> {

    private final FSDataOutputStream output;

    public CustomRecordWriter(FSDataOutputStream output) {
      this.output = output;
    }

    @Override
    public void write(Text key, CustomWritable value) throws IOException, InterruptedException {
      output.write((key.toString() + "\t" + value.toString()).getBytes());
      output.write("\n".getBytes());
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
      if (output != null) {
        output.close();
      }
    }
  }
}
